package main.java.blackList

import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.SparkSession
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
  * BlackListIPDemo
  *
  * @author zhangyimin
  *  2018-11-15 下午10:54
  * @version 1.0
  */
object BlackListIPDemo {


  def main(args: Array[String]): Unit = {
    //        1,201.105.101.102,http://mystore.jsp/?productid=1,2017020029,2,1
    Logger.getLogger("org.apache.spark").setLevel(Level.ERROR)
    Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF)
    val sparkSession=SparkSession.builder().master("local[2]").appName("black_list_spark").getOrCreate()
//一个streaming中只能有一个SparkContext对象
    val sc=sparkSession.sparkContext

    //                                               采样率10S
    val scc=new StreamingContext(sc,batchDuration = Seconds(10))






    val sqlContext=sparkSession.sqlContext
    val kafkaStreams=KafkaUtils.createStream(
      scc,
      "10.16.7.36:2181",
      "top_ip",
      Map("top_ip"->1),
      StorageLevel.MEMORY_ONLY_SER_2)
    val logRdd=kafkaStreams.map(_._2)
    val splitRdd=logRdd.filter(_.split(",").length==6).map(_.split(",")).map(x=>(x(0),1))
      //                                         窗口长度 30S  滑动距离 10S
//      窗口长度和采样率  成整数倍
      .reduceByKeyAndWindow((a:Int,b:Int)=>(a+b),Seconds(30),Seconds(10))

    import sqlContext.implicits._

    val res=splitRdd.filter(_._2>5).foreachRDD(rdd=>{
      rdd.map(x=>new HotIp(x._1,x._2)).toDF().createTempView("HotIp")



      sparkSession.sql("select * from  HotIp")




    })
    scc.start()
    scc.awaitTermination()




//      userInfo.createTempView("user_info")
//
//      val countDF=sparkSession.sql("select user_id,ip,count(ip) count_ip from user_info group by user_id,ip")
//
//      val props = new Properties()
//      //    props.setProperty("url","jdbc:mysql://localhost:3306/hive_etl?characterEncoding=utf-8&useSSL=false")
//      props.setProperty("user", "root")
//      props.setProperty("password", "123456")
//
//
//
//      countDF.map(x=>new HotIp(x.getString(0),x.getString(1),x.getLong(2))).map(x=>{
//        if(x.pv>4){
//          x.tpDF().write.mode("append").jdbc("jdbc:mysql://localhost:3306/hive_etl?characterEncoding=utf-8&useSSL=false", "stu", props)
//        }
//      })
  }
  case class HotIp(user_id:String,pv:Long)

  case class UserInfo(user_id:String,user_name:String)

}
